package com.awspaas.user.apps.cpq.job;

import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

import com.actionsoft.bpms.schedule.IJob;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.*;

public class RabbitMQGetMsgJob implements IJob{
	private final static String QUEUE_NAME = "test_queue";
	@Override
	public void execute(JobExecutionContext arg0) throws JobExecutionException {
		try {
			ConnectionFactory factory = new ConnectionFactory();
	        factory.setHost("localhost");
	        Connection connection = factory.newConnection();
	        Channel channel = connection.createChannel();
	        
	        try {
	            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
	            channel.basicQos(1);
	            System.out.println(" [*] Waiting for messages...");

	            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
	                try {
	                    String message = new String(delivery.getBody(), "UTF-8");
	                    System.out.println(" [x] Received '" + message + "'");
	                 
	                    // 确保通道未关闭再确认
	                    if(channel.isOpen()) {
	                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
	                        System.out.println(" [√] Message processed");
	                    }
	                } catch (Exception e) {
	                    System.err.println(" [×] Process failed: " + e.getMessage());
	                    if(channel.isOpen()) {
	                        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
	                    }
	                }
	            };

	            channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
	            // 保持主线程运行
	            while(true) {
	                Thread.sleep(1000);
	            }
	        } finally {
	            if(channel != null && channel.isOpen()) channel.close();
	            if(connection != null && connection.isOpen()) connection.close();
	        }
		} catch (Exception e2) {
			
		}
		   
	    }
	}